package io.github.melin.sqlflow.parser.flink

import io.github.melin.sqlflow.analyzer.Analysis
import io.github.melin.sqlflow.analyzer.StatementAnalyzer
import io.github.melin.sqlflow.metadata.SimpleMetadataService
import io.github.melin.sqlflow.metadata.SqlMetadataExtractor
import io.github.melin.sqlflow.parser.SqlFlowParser
import io.github.melin.sqlflow.parser.hive.HiveSqlLineageTest
import io.github.melin.sqlflow.tree.statement.EmptyStatement
import io.github.melin.sqlflow.util.JsonUtils
import io.github.melin.superior.common.relational.create.CreateTableAsSelect
import io.github.melin.superior.common.relational.dml.InsertTable
import io.github.melin.superior.common.relational.dml.MergeTable
import io.github.melin.superior.parser.flink.FlinkSqlHelper
import org.apache.commons.lang3.StringUtils
import org.junit.Test
import java.io.IOException
import java.util.*

class FlinkSqlLineageTest3 {

    protected val SQL_PARSER = SqlFlowParser()

    @Test
    @Throws(Exception::class)
    fun testInsertInto() {
        val script = """              
           INSERT overwrite TABLE ads_trans_elecric_w
           SELECT  t1.prov,
                current_timestamp() data_store_time,
                regexp_replace(t2.eltct_desc, 'sdf','') eltct_desc
           from dws_eltct_pro_w t1
           LEFT JOIN dwd_pub_fill_w t2
               t2 on t1.pro_id = t2.pro_id
           and t1.week_end = t2.week_end

        """.trimIndent()

        val statement = SQL_PARSER.createStatement(script)
        val analysis = Analysis(statement, emptyMap())
        val statementAnalyzer = StatementAnalyzer(
            analysis,
            SimpleFlinkMetadataService(), SQL_PARSER
        )
        statementAnalyzer.analyze(statement, Optional.empty())

        System.out.println(JsonUtils.toJSONString(analysis.getTarget().get()));
    }

    /**
     * 测试函数
     *
     * @throws IOException 如果发生IO异常
     */
    @Test
    @Throws(IOException::class)
    fun test4() {
        val script=createsql();
        println(script)

        val sqlList = parseAndFilterSqlStatements(script)

        val metadataService = SimpleMetadataService("default")
        val multipleStatement = SQL_PARSER.createStatements(sqlList)
        for (statement in multipleStatement) {
            val extractor = SqlMetadataExtractor()
            extractor.clearContext()
            extractor.process(statement)
            printSourceTableInfo(extractor)
            printTableAliasMapping(extractor)
            printLastSelectColumns(extractor)
            for (table in extractor.tables) {
                metadataService.addTableMetadata(table)
            }
            extractor.removeContext()
        }


        // 创建StatementAnalyzer实例
        val statementAnalyzer = StatementAnalyzer(
            Analysis(EmptyStatement(), emptyMap()),  // 创建一个临时Analysis对象
            metadataService,
            SQL_PARSER
        )


        // 使用analyzeMultiple方法处理多个语句
        val results = statementAnalyzer.analyzeMultiple(multipleStatement, Optional.empty())

        // 处理分析结果
        for (result in results) {
            println("+++++++++++++++++++++")
            println(result)
        }
    }


    /**
     * 解析并过滤SQL语句
     *
     * @param script SQL脚本
     * @return 符合条件的SQL语句列表
     */
    private fun createsql(): String{
        val script = """
        insert  overwrite  dwd_wind_fan_wind_resrc partition (dt='1231')
        SELECT  t1.fan_num                                              fan_num               --风机编号
             ,t2.fan_cap                                             fan_type              --风机类型
             ,t1.msr_point_colt_time                                 msr_point_colt_time   --测点采集时间
             ,t1.sec_wind_speed+t1.fan_num                                       sec_wind_speed        --秒级风速
             ,substr(current_timestamp(),1,19)                       data_store_time       --数据入库时间
        from
            (
                SELECT
                    case
                        when length(ti.desc) = 8 then concat('#0',SUBSTRING(ti.desc,1,1))
                        when length(ti.desc) = 9 then concat('#',SUBSTRING(ti.desc,1,2))
                        ELSE NULL
                        end as fan_num                --风机编号
                     ,ts   msr_point_colt_time     --测点采集时间
                     ,r32  sec_wind_speed          --秒级风速
                from        ods_point_st_telemetry_short_float as tsf
                                left join   dim_pub_fan_telemetry_info ti
                                            on tsf.ip = ti.ip
                where       ti.desc LIKE '%风机瞬时风速%'   --只取瞬时风速数据
                  and tsf.r32 BETWEEN 0 and 50             --对风速大于50m/s的数据则进行去除
                  and dt=from_unixtime(unix_timestamp('1231','yyyymmdd'),'yyyy-mm-dd')
            )t1
                left join   dim_pub_fan_info t2
                            on t1.fan_num = t2.fan_num
        ;
    """.trimIndent()
        return script
    }

    /**
     * 解析并过滤SQL语句
     *
     * @param script SQL脚本
     * @return 符合条件的SQL语句列表
     */
    private fun parseAndFilterSqlStatements(script: String): MutableList<String> {
        val statements = FlinkSqlHelper.parseMultiStatement(script)
        val sqlList = mutableListOf<String>()
        for (statement in statements) {
            if (statement is InsertTable || statement is CreateTableAsSelect || statement is MergeTable) {
                if (StringUtils.isNotBlank(statement.getSql())) {
                    sqlList.add(statement.getSql())
                }
            }
        }
        return sqlList
    }

    /**
     * 打印源表信息
     *
     * @param extractor SqlMetadataExtractor实例
     */
    private fun printSourceTableInfo(extractor: SqlMetadataExtractor) {
        println("\n源表信息:")
        for (table in extractor.sourceTables) {
            println("表名: ${table.tableName}")
            println("列: ${table.columns.joinToString(", ")}")
        }
    }

    /**
     * 打印表别名映射
     *
     * @param extractor SqlMetadataExtractor实例
     */
    private fun printTableAliasMapping(extractor: SqlMetadataExtractor) {
        println("\n表别名映射:")
        for ((alias, realTable) in extractor.tableAliases) {
            println("$alias -> $realTable")
        }
    }

    /**
     * 打印最后的SELECT列
     *
     * @param extractor SqlMetadataExtractor实例
     */
    private fun printLastSelectColumns(extractor: SqlMetadataExtractor) {
        println("\n最后的SELECT列:")
        println(extractor.lastSelectColumns.joinToString(", "))
    }

}